package cn.hxth.datasync.mq;


import cn.hxth.api.model.Consumer;
import cn.hxth.api.model.ProductBaseInfo;
import cn.hxth.api.model.Remind;
import com.alibaba.fastjson.JSONObject;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;

import javax.annotation.Resource;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;

@Component  
@RabbitListener(queues = "data-change-queue")  
public class DataChangeQueueReceiver {
	@Resource
	private HttpHeaders headers ;
	@Resource(name = "myRedisTemplate")
	private RedisTemplate<String,Object> redisTemplate ;
	@Resource
	private RestTemplate restTemplate ;
	@Autowired
	private RabbitMQSender rabbitMQSender;
	
	private Set<String> dimDataChangeMessageSet = 
			Collections.synchronizedSet(new HashSet<String>()); 

	
	public DataChangeQueueReceiver() {
		new SendThread().start();
	}
	
    @RabbitHandler  
    public void process(String message) {
    	// 对这个message进行解析
    	JSONObject jsonObject = JSONObject.parseObject(message);
    	
    	// 先获取data_type
    	String dataType = jsonObject.getString("data_type");  
    	if("plant".equals(dataType)) {
    		processPlantDataChangeMessage(jsonObject);
    	} else if("consumer".equals(dataType)) {
    		processConsumerDataChangeMessage(jsonObject);
    	} else if("product_spec".equals(dataType)) {
    		processProductSpecDataChangeMessage(jsonObject);
    	} else if("product_name".equals(dataType)) {
			processProductNameDataChangeMessage(jsonObject);
     	} else if("product".equals(dataType)) {
			processProductDataChangeMessage(jsonObject);
		}else if("remind".equals(dataType)) {
				processRemindDataChangeMessage(jsonObject);
     	}else if(dataType==null){
			processDatabaseChangeMessage(jsonObject);
    		/* else if("product_specification".equals(dataType)) {
     		processProductSpecificationDataChangeMessage(jsonObject);  
     	}*/
		}
    }



	private void processPlantDataChangeMessage(JSONObject messageJSONObject) {
    	Long id = messageJSONObject.getLong("cid");
    	String eventType = messageJSONObject.getString("event_type"); 
    	
    	if("add".equals(eventType) || "update".equals(eventType)) {
			JSONObject plantObject = JSONObject.parseObject(restTemplate.exchange("http://localhost:81/pages/plant/getPlantForMq?cid="+id,HttpMethod.GET,new HttpEntity<>(headers),String.class).getBody());
    				this.redisTemplate.opsForValue().set("plant_" + id, plantObject.toJSONString());

    			}
		else if ("delete".equals(eventType)) {

			this.redisTemplate.delete("plant_" + id);
		}
		dimDataChangeMessageSet.add("{\"dim_type\": \"plant\", \"cid\": " + id + "}");

    	}

    
    private void processConsumerDataChangeMessage(JSONObject messageJSONObject) {
    	Long id = messageJSONObject.getLong("cid");
    	String eventType = messageJSONObject.getString("event_type"); 
    	
    	if("add".equals(eventType) || "update".equals(eventType)) {
    		long startTime = System.currentTimeMillis() ;
    		long waitTime = 0l ;
    		long endTime = 0l ;
			JSONObject dataJSONObject=null ;

    		while (waitTime<2000){
				dataJSONObject = JSONObject.parseObject(restTemplate.exchange("http://localhost:81/pages/consumer/getConsumerForMq?cid="+id,HttpMethod.GET,new HttpEntity<>(headers), String.class).getBody());
				if(dataJSONObject!=null){
					break;
				}
    			System.out.println("************************");
				try {
					Thread.sleep(200);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				endTime = System.currentTimeMillis() ;
				waitTime = endTime - startTime ;
			}
System.out.println("需方信息:"+dataJSONObject);
    		this.redisTemplate.opsForValue().set("consumer_" + id, dataJSONObject.toJSONString());
    	} else if ("delete".equals(eventType)) {
    		this.redisTemplate.delete("consumer_" + id);
    	}
    	  
    	dimDataChangeMessageSet.add("{\"dim_type\": \"consumer\", \"cid\": " + id + "}");
    }
    
    private void processProductSpecDataChangeMessage(JSONObject messageJSONObject) {
    	Long id = messageJSONObject.getLong("sid");
    	int nid = messageJSONObject.getInteger("nid") ;
    	String eventType = messageJSONObject.getString("event_type"); 
    	
    	if("add".equals(eventType)) {
    		JSONObject dataJSONObject = JSONObject.parseObject(restTemplate.exchange("http://localhost:81/pages/spec/getProductSpecForMq?sid="+id,HttpMethod.GET,new HttpEntity<>(headers), String.class).getBody());
    		this.redisTemplate.opsForHash().put("specList_"+nid,"spec_"+id,dataJSONObject.toJSONString());
    		this.redisTemplate.opsForValue().set("spec_"+id,dataJSONObject.toJSONString()) ;
    	} else if ("delete".equals(eventType)) {
			this.redisTemplate.opsForHash().delete("specList_"+nid,"spec_"+id);
			this.redisTemplate.delete("spec_"+id);
    	}
    	  
    	dimDataChangeMessageSet.add("{\"dim_type\": \"product_spec\", \"sid\": " + id + "}");
    }
	private void processProductNameDataChangeMessage(JSONObject messageJSONObject) {
		Long id = messageJSONObject.getLong("nid");
		String eventType = messageJSONObject.getString("event_type");

		if("add".equals(eventType)) {
			JSONObject dataJSONObject = JSONObject.parseObject(restTemplate.exchange("http://localhost:81/pages/productName/getProductNameForMq?nid="+id,HttpMethod.GET,new HttpEntity<>(headers), String.class).getBody());
			this.redisTemplate.opsForHash().put("nameList","name_"+id,dataJSONObject.toJSONString());
		} else if ("delete".equals(eventType)) {
			this.redisTemplate.opsForHash().delete("nameList","name_"+id);
		}

		dimDataChangeMessageSet.add("{\"dim_type\": \"product_name\", \"nid\": " + id + "}");
	}
    
    private void processProductDataChangeMessage(JSONObject messageJSONObject) {
    	Long id = messageJSONObject.getLong("pid");
    	String eventType = messageJSONObject.getString("event_type");

    	if("add".equals(eventType) || "update".equals(eventType)) {

    		JSONObject dataJSONObject = JSONObject.parseObject(restTemplate.exchange("http://localhost:81/pages/product/getProductForUpdate?pid="+id, HttpMethod.GET,new HttpEntity<>(headers), String.class).getBody());
			if("add".equals(eventType)){
				long pid = dataJSONObject.getLong("pid");
				int sid = dataJSONObject.getJSONObject("spec").getInteger("sid") ;
				this.redisTemplate.opsForValue().set(sid+"-product",String.valueOf(pid));
			}
    		this.redisTemplate.opsForValue().set("product_" + id, dataJSONObject.toJSONString());
    	} else if ("delete".equals(eventType)) {

			Integer sid  = messageJSONObject.getInteger("sid");
			System.out.println(sid);
			this.redisTemplate.delete("product_" + id);
    		this.redisTemplate.delete("product_base_info_"+id);
    		this.redisTemplate.delete(sid+"-product");
    	}
    	  
    	dimDataChangeMessageSet.add("{\"dim_type\": \"product\", \"pid\": " + id + "}");
    }

    
   /* private void processProductPropertyDataChangeMessage(JSONObject messageJSONObject) {
    	Long id = messageJSONObject.getLong("id"); 
    	Long productId = messageJSONObject.getLong("product_id");
    	String eventType = messageJSONObject.getString("event_type"); 
    	
    	if("add".equals(eventType) || "update".equals(eventType)) { 
    		JSONObject dataJSONObject = JSONObject.parseObject (restTemplate.getForObject("http://localhost:8767/productProperty/findById?id="+id, String.class));
    		Jedis jedis = jedisPool.getResource();
    		jedis.set("product_property_" + productId, dataJSONObject.toJSONString());
    	} else if ("delete".equals(eventType)) {
    		Jedis jedis = jedisPool.getResource();
    		jedis.del("product_property_" + productId);
    	}
    	  
    	dimDataChangeMessageSet.add("{\"dim_type\": \"product\", \"id\": " + productId + "}");
    }
    
    private void processProductSpecificationDataChangeMessage(JSONObject messageJSONObject) {
    	Long id = messageJSONObject.getLong("id"); 
    	Long productId = messageJSONObject.getLong("product_id");
    	String eventType = messageJSONObject.getString("event_type"); 
    	
    	if("add".equals(eventType) || "update".equals(eventType)) { 
    		JSONObject dataJSONObject = JSONObject.parseObject(restTemplate.getForObject("http://localhost:8767/productSpecification/findById?id="+id, String.class));
    		Jedis jedis = jedisPool.getResource();
    		jedis.set("product_specification_" + productId, dataJSONObject.toJSONString());
    	} else if ("delete".equals(eventType)) {
    		Jedis jedis = jedisPool.getResource();
    		jedis.del("product_specification_" + productId);
    	}
    	  
    	dimDataChangeMessageSet.add("{\"dim_type\": \"product\", \"id\": " + productId + "}");
    }*/
   private void processRemindDataChangeMessage(JSONObject messageJSONObject) {
	   Long id = messageJSONObject.getLong("sid");
	   //int nid = messageJSONObject.getInteger("nid") ;
	   String eventType = messageJSONObject.getString("event_type");

	   if("add".equals(eventType)||"update".equals(eventType)) {
		   JSONObject dataJSONObject = JSONObject.parseObject(restTemplate.exchange("http://localhost:81/pages/remind/getRemindForMq?sid="+id,HttpMethod.GET,new HttpEntity<>(headers) ,String.class).getBody());
		   this.redisTemplate.opsForHash().put("remindList","remind_"+id,dataJSONObject.toJSONString());
	   } else if ("delete".equals(eventType)) {
		   this.redisTemplate.opsForHash().delete("remindList","remind_"+id);
	   }

	   dimDataChangeMessageSet.add("{\"dim_type\": \"remind\", \"sid\": " + id + "}");
   }
	private void processDatabaseChangeMessage(JSONObject messageJSONObject) {
		String eventType = messageJSONObject.getString("event_type");
		if(eventType.equals("updateProductBaseInfo")){
			ProductBaseInfo productBaseInfo = JSONObject.parseObject(messageJSONObject.getString("data"), ProductBaseInfo.class);
			this.restTemplate.postForObject("http://localhost:81/pages/product/updateProductBaseInfo", new HttpEntity<>(productBaseInfo,headers),Boolean.class);
			dimDataChangeMessageSet.add("{\"dim_type\": \"product\", \"pid\": " + productBaseInfo.getPid() + "}");
			//this.rabbitMQSender.send("aggr-data-change-queue", "{\"dim_type\": \"product\", \"pid\": " + productBaseInfo.getPid() + "}");

		}else if(eventType.equals("updatePlantTime")){
			int cid = messageJSONObject.getInteger("data") ;
			this.restTemplate.exchange("http://localhost:81/pages/plant/updatePlantInstockRecordTime?cid="+cid,HttpMethod.POST,new HttpEntity<>(headers),Boolean.class) ;
		}else if(eventType.equals("insertRemind")){

			Remind remind = JSONObject.parseObject(messageJSONObject.getString("data"),Remind.class) ;
			/*while ((true)){
				System.out.println(messageJSONObject.toJSONString(remind));
				if(1!=1){
					break;
				}
			}*/
			this.restTemplate.postForObject("http://localhost:81/pages/remind/insertRemind",new HttpEntity<>(remind,headers),Boolean.class) ;
		}
		else if(eventType.equals("updateRemindCount")){
			Remind remind = JSONObject.parseObject(messageJSONObject.getString("data"),Remind.class) ;
			this.restTemplate.postForObject("http://localhost:81/pages/remind/updateRemindCount",new HttpEntity<>(remind,headers),Boolean.class) ;
		}else if(eventType.equals("deleteRemind")){
			int sid =messageJSONObject.getInteger("data") ;
			this.restTemplate.exchange("http://localhost:81/pages/remind/deleteRemind?sid="+sid,HttpMethod.POST,new HttpEntity<>(headers),Boolean.class) ;
		}else if(eventType.equals("updateTotal")){
			Consumer consumer = JSONObject.parseObject(messageJSONObject.getString("data"), Consumer.class);
			this.restTemplate.postForObject("http://localhost:81/pages/consumer/updateTotal",new HttpEntity<>(consumer,headers),Boolean.class) ;
		}
	}
    
    private class SendThread extends Thread {
    	
    	@Override
    	public void run() {
    		while(true) {
    			if(!dimDataChangeMessageSet.isEmpty()) {
    				for(String message : dimDataChangeMessageSet) {
    					rabbitMQSender.send("aggr-data-change-queue", message);   
    				}
    				dimDataChangeMessageSet.clear();
    			}
    			try {
					Thread.sleep(100);
				} catch (InterruptedException e) {
					e.printStackTrace();
				} 
    		}
    	}
    	
    }
  
}  